package com.github.binarylei.wordcount.topology;

import com.github.binarylei.wordcount.bolt.WorldCountBolt;
import com.github.binarylei.wordcount.bolt.WorldReportBolt;
import com.github.binarylei.wordcount.bolt.WorldSplitBolt;
import com.github.binarylei.wordcount.spout.WorldSpout;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

/**
 * @author: leigang
 * @version: 2018-04-07
 */
public class WordTopology {

    public static void main(String[] args) throws InterruptedException {

        WorldSpout spout = new WorldSpout();
        WorldSplitBolt splitBolt = new WorldSplitBolt();
        WorldCountBolt countBolt = new WorldCountBolt();
        WorldReportBolt reportBolt = new WorldReportBolt();

        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("spout", spout);
        builder.setBolt("splitBolt", splitBolt, 5).shuffleGrouping("spout");
        builder.setBolt("countBolt", countBolt, 5)
                .fieldsGrouping("splitBolt", new Fields("word"));
        builder.setBolt("reportBolt", reportBolt).globalGrouping("countBolt");

        Config cfg = new Config();
        //cfg.setNumWorkers(1);
        cfg.setDebug(true);

        //1. 本地模式
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("topo", cfg, builder.createTopology());
        Thread.sleep(20 * 1000);
        cluster.killTopology("topo");
        cluster.shutdown();

        //2. 集群模式
        //cfg.setNumWorkers(1);
        //cfg.setDebug(true);
        //StormSubmitter.submitTopology("top1", cfg, builder.createTopology());

    }
}
